package com.wande.dataplatform.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.domain.DataSource;
import com.wande.dataplatform.domain.DevInstance;
import com.wande.dataplatform.domain.DevTask;
import com.wande.dataplatform.domain.bo.DevTaskBo;
import com.wande.dataplatform.domain.dto.ExecutionPlanDTO;
import com.wande.dataplatform.domain.vo.DevInstanceVO;
import com.wande.dataplatform.domain.vo.DevTaskVO;
import com.wande.dataplatform.mapper.DevInstanceMapper;
import com.wande.dataplatform.mapper.DevTaskMapper;
import com.wande.dataplatform.service.IDataDevService;
import com.wande.dataplatform.service.IDataSourceService;
import com.wande.dataplatform.service.IDolphinSchedulerSyncService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.*;
import java.time.LocalDateTime;
import java.util.*;

/**
 * 数据开发服务实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DataDevServiceImpl implements IDataDevService {

    private final DevTaskMapper devTaskMapper;
    private final DevInstanceMapper devInstanceMapper;
    private final IDataSourceService dataSourceService;
    private final IDolphinSchedulerSyncService dolphinSchedulerSyncService;

    @Override
    public TableDataInfo<DevTaskVO> queryPageList(DevTaskBo bo, PageQuery pageQuery) {
        LambdaQueryWrapper<DevTask> wrapper = buildQueryWrapper(bo);
        Page<DevTaskVO> page = devTaskMapper.selectVoPage(pageQuery.build(), wrapper);
        return TableDataInfo.build(page);
    }

    @Override
    public DevTaskVO queryById(Long id) {
        DevTaskVO vo = devTaskMapper.selectVoById(id);
        if (vo == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }
        return vo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long createTask(DevTaskBo bo) {
        DevTask task = BeanUtil.toBean(bo, DevTask.class);
        task.setStatus(0); // 草稿状态
        devTaskMapper.insert(task);
        return task.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateTask(DevTaskBo bo) {
        DevTask task = devTaskMapper.selectById(bo.getId());
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }
        
        BeanUtil.copyProperties(bo, task);
        return devTaskMapper.updateById(task) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteTask(Long id) {
        DevTask task = devTaskMapper.selectById(id);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }
        
        // 如果已发布，需要先下线
        if (task.getStatus() == 1) {
            throw new ServiceException("任务已发布，请先下线后再删除");
        }
        
        return devTaskMapper.deleteById(id) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Map<String, Object> executeSql(Long taskId, Map<String, Object> params) {
        DevTask task = devTaskMapper.selectById(taskId);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }

        // 创建执行实例
        DevInstance instance = new DevInstance();
        instance.setTaskId(taskId);
        instance.setStatus("RUNNING");
        instance.setStartTime(LocalDateTime.now());
        devInstanceMapper.insert(instance);

        Map<String, Object> result = new HashMap<>();
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            // 获取数据源连接
            conn = dataSourceService.getConnection(task.getDatasourceId());
            stmt = conn.createStatement();

            // 执行SQL
            String sql = task.getContent();
            boolean hasResultSet = stmt.execute(sql);

            if (hasResultSet) {
                // 查询语句
                rs = stmt.getResultSet();
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();

                // 构建列信息
                List<Map<String, Object>> columns = new ArrayList<>();
                for (int i = 1; i <= columnCount; i++) {
                    Map<String, Object> column = new HashMap<>();
                    column.put("name", metaData.getColumnName(i));
                    column.put("type", metaData.getColumnTypeName(i));
                    columns.add(column);
                }

                // 获取数据
                List<Map<String, Object>> rows = new ArrayList<>();
                int rowCount = 0;
                while (rs.next() && rowCount < 1000) { // 限制最多返回1000行
                    Map<String, Object> row = new HashMap<>();
                    for (int i = 1; i <= columnCount; i++) {
                        row.put(metaData.getColumnName(i), rs.getObject(i));
                    }
                    rows.add(row);
                    rowCount++;
                }

                result.put("columns", columns);
                result.put("rows", rows);
                result.put("rowCount", rowCount);

                // 更新实例
                instance.setResultRows(rowCount);
            } else {
                // 更新/删除/插入语句
                int updateCount = stmt.getUpdateCount();
                result.put("affectedRows", updateCount);
                instance.setResultRows(updateCount);
            }

            // 更新实例状态
            instance.setStatus("SUCCESS");
            instance.setEndTime(LocalDateTime.now());
            instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
            devInstanceMapper.updateById(instance);

            result.put("instanceId", instance.getId());
            result.put("status", "SUCCESS");

        } catch (Exception e) {
            log.error("执行SQL失败", e);
            
            // 更新实例状态
            instance.setStatus("FAILED");
            instance.setEndTime(LocalDateTime.now());
            instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
            instance.setErrorMsg(e.getMessage());
            devInstanceMapper.updateById(instance);

            result.put("instanceId", instance.getId());
            result.put("status", "FAILED");
            result.put("error", e.getMessage());
            
            throw new ServiceException("SQL执行失败: " + e.getMessage());
        } finally {
            closeResources(rs, stmt, conn);
        }

        return result;
    }

    @Override
    public Map<String, Object> executeScript(Long taskId, Map<String, Object> params) {
        DevTask task = devTaskMapper.selectById(taskId);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }

        // 创建执行实例
        DevInstance instance = new DevInstance();
        instance.setTaskId(taskId);
        instance.setStatus("RUNNING");
        instance.setStartTime(LocalDateTime.now());
        devInstanceMapper.insert(instance);

        Map<String, Object> result = new HashMap<>();

        try {
            // 这里应该集成脚本执行引擎（如Python、Shell等）
            // 简化实现，仅记录日志
            log.info("执行脚本任务: {}, 类型: {}", task.getName(), task.getTaskType());

            // 模拟执行
            Thread.sleep(1000);

            // 更新实例状态
            instance.setStatus("SUCCESS");
            instance.setEndTime(LocalDateTime.now());
            instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
            instance.setLogs("脚本执行成功");
            devInstanceMapper.updateById(instance);

            result.put("instanceId", instance.getId());
            result.put("status", "SUCCESS");
            result.put("message", "脚本执行成功");

        } catch (Exception e) {
            log.error("执行脚本失败", e);
            
            // 更新实例状态
            instance.setStatus("FAILED");
            instance.setEndTime(LocalDateTime.now());
            instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
            instance.setErrorMsg(e.getMessage());
            devInstanceMapper.updateById(instance);

            result.put("instanceId", instance.getId());
            result.put("status", "FAILED");
            result.put("error", e.getMessage());
            
            throw new ServiceException("脚本执行失败: " + e.getMessage());
        }

        return result;
    }

    @Override
    public Boolean stopExecution(Long instanceId) {
        DevInstance instance = devInstanceMapper.selectById(instanceId);
        if (instance == null) {
            throw new ServiceException("实例不存在");
        }

        if (!"RUNNING".equals(instance.getStatus())) {
            throw new ServiceException("实例未在运行中");
        }

        // 更新实例状态
        instance.setStatus("STOPPED");
        instance.setEndTime(LocalDateTime.now());
        instance.setDuration(java.time.Duration.between(instance.getStartTime(), instance.getEndTime()).toMillis());
        
        return devInstanceMapper.updateById(instance) > 0;
    }

    @Override
    public ExecutionPlanDTO getExecutionPlan(Long taskId) {
        DevTask task = devTaskMapper.selectById(taskId);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }

        if (!"SQL".equals(task.getTaskType())) {
            throw new ServiceException("只有SQL任务支持执行计划分析");
        }

        ExecutionPlanDTO plan = new ExecutionPlanDTO();
        Connection conn = null;
        Statement stmt = null;
        ResultSet rs = null;

        try {
            conn = dataSourceService.getConnection(task.getDatasourceId());
            stmt = conn.createStatement();

            // 执行EXPLAIN获取执行计划
            String explainSql = "EXPLAIN " + task.getContent();
            rs = stmt.executeQuery(explainSql);

            StringBuilder planText = new StringBuilder();
            List<ExecutionPlanDTO.ExecutionStep> steps = new ArrayList<>();
            int stepId = 0;

            while (rs.next()) {
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();
                
                for (int i = 1; i <= columnCount; i++) {
                    planText.append(rs.getString(i)).append(" ");
                }
                planText.append("\n");

                // 解析执行步骤（简化实现）
                ExecutionPlanDTO.ExecutionStep step = new ExecutionPlanDTO.ExecutionStep();
                step.setStepId(++stepId);
                step.setOperationType(rs.getString(1));
                steps.add(step);
            }

            plan.setPlanText(planText.toString());
            plan.setSteps(steps);

        } catch (Exception e) {
            log.error("获取执行计划失败", e);
            throw new ServiceException("获取执行计划失败: " + e.getMessage());
        } finally {
            closeResources(rs, stmt, conn);
        }

        return plan;
    }

    @Override
    public TableDataInfo<DevInstanceVO> queryInstanceList(Long taskId, PageQuery pageQuery) {
        LambdaQueryWrapper<DevInstance> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(DevInstance::getTaskId, taskId);
        wrapper.orderByDesc(DevInstance::getCreateTime);
        
        Page<DevInstanceVO> page = devInstanceMapper.selectVoPage(pageQuery.build(), wrapper);
        return TableDataInfo.build(page);
    }

    @Override
    public DevInstanceVO queryInstanceById(Long instanceId) {
        DevInstanceVO vo = devInstanceMapper.selectVoById(instanceId);
        if (vo == null) {
            throw new ServiceException("实例不存在");
        }
        return vo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean publishTask(Long taskId) {
        DevTask task = devTaskMapper.selectById(taskId);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }

        if (task.getStatus() == 1) {
            throw new ServiceException("任务已发布");
        }

        // 如果配置了调度，创建DolphinScheduler工作流
        if ("CRON".equals(task.getScheduleType()) || "DEPENDENCY".equals(task.getScheduleType())) {
            // 这里应该调用DolphinScheduler API创建工作流
            log.info("创建DolphinScheduler工作流: {}", task.getName());
        }

        task.setStatus(1);
        return devTaskMapper.updateById(task) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean offlineTask(Long taskId) {
        DevTask task = devTaskMapper.selectById(taskId);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.DEV_TASK_NOT_FOUND.getMessage());
        }

        if (task.getStatus() != 1) {
            throw new ServiceException("任务未发布");
        }

        // 如果有DolphinScheduler工作流，需要停止
        if (task.getDsWorkflowId() != null) {
            log.info("停止DolphinScheduler工作流: {}", task.getDsWorkflowId());
        }

        task.setStatus(2);
        return devTaskMapper.updateById(task) > 0;
    }

    /**
     * 构建查询条件
     */
    private LambdaQueryWrapper<DevTask> buildQueryWrapper(DevTaskBo bo) {
        LambdaQueryWrapper<DevTask> wrapper = Wrappers.lambdaQuery();
        wrapper.like(ObjectUtil.isNotEmpty(bo.getName()), DevTask::getName, bo.getName());
        wrapper.eq(ObjectUtil.isNotEmpty(bo.getTaskType()), DevTask::getTaskType, bo.getTaskType());
        wrapper.eq(ObjectUtil.isNotEmpty(bo.getStatus()), DevTask::getStatus, bo.getStatus());
        wrapper.eq(ObjectUtil.isNotEmpty(bo.getProjectId()), DevTask::getProjectId, bo.getProjectId());
        wrapper.orderByDesc(DevTask::getCreateTime);
        return wrapper;
    }

    /**
     * 关闭资源
     */
    private void closeResources(ResultSet rs, Statement stmt, Connection conn) {
        try {
            if (rs != null) rs.close();
            if (stmt != null) stmt.close();
            if (conn != null) conn.close();
        } catch (SQLException e) {
            log.error("关闭资源失败", e);
        }
    }
}
